package com.fy.service.server.consumer;

import com.fy.service.server.listener.MessagePushListener;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.net.InetAddress;
import java.net.UnknownHostException;

/**
 * Description：消费者配置,负责消费其他节点传递过来的message
 */
@Configuration
public class MessagePushConsumer {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private MessagePushListener messagePushListener;

    @Value("${spring.rocketmq.consumer.namesrvAddr}")
    private String namesrvAddr;

    @Value("${spring.rocketmq.consumer.groupName}")
    private String groupName;

    @Value("${spring.rocketmq.consumer.consumeThreadMin}")
    private int consumeThreadMin;

    @Value("${spring.rocketmq.consumer.consumeThreadMax}")
    private int consumeThreadMax;
    @Value("${netty.server.cer.port}")
    private int nettyProt;
    @Value("${spring.cloud.nacos.discovery.ip}")
    private String host;



    @Bean(initMethod = "start", destroyMethod = "shutdown",name = "messageMQPushConsumer")
    public DefaultMQPushConsumer getRocketMQConsumer() {
        try {
            String hostAddress = host;
            String topic = (hostAddress + ":" + nettyProt).replace(".","_").replace(":","_");
            //设置唯一的group 否则消息消费会在同一个group下负载均衡
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(topic);
            consumer.setNamesrvAddr(namesrvAddr);
            consumer.setConsumeThreadMin(consumeThreadMin);
            consumer.setConsumeThreadMax(consumeThreadMax);
            //设置从队列头部开始消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //只消费一次
            consumer.setMessageModel(MessageModel.CLUSTERING);
            consumer.setVipChannelEnabled(false);
            //设置一次性消费多少条
            consumer.setPullBatchSize(100);
            consumer.registerMessageListener(messagePushListener);
            //消费目标为本机ip target 为oneOnOneMessage
            consumer.subscribe(topic,"");
            //consumer.start();
            logger.info("================>下发 消费者创建完成，ConsumerGroupName{}<================",groupName);
            return consumer;
        } catch (Exception  e) {
            logger.error("消费者启动失败:" ,e);
        }
        return null;
    }

}
